package com;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class Hbase {
	private static Configuration config = null;
	/**
	 * 初始化配置
	 */
	static {
		config = HBaseConfiguration.create();
	}

	public void selectByFilter(String tablename, List<String[]> arr)
			throws IOException {
		HTable table = new HTable(config, tablename);
		FilterList filterList = new FilterList();
		Scan s1 = new Scan();
		for (String[] s : arr) { // 各个条件之间是“与”的关系
			filterList.addFilter(new SingleColumnValueFilter(Bytes
					.toBytes(s[0]), Bytes.toBytes(s[1]), CompareOp.EQUAL, Bytes
					.toBytes(s[2])));
			// 添加下面这一行后，则只返回指定的cell，同一行中的其他cell不返回
			// s1.addColumn(Bytes.toBytes(s[0]), Bytes.toBytes(s[1]));
		}
		s1.setFilter(filterList);
		ResultScanner ResultScannerFilterList = table.getScanner(s1);
		for (Result r : ResultScannerFilterList) {
			byte[] data = r.getRow();
			int offset = 0;
			int length = 0;
			String datas = new String(data);
			System.out.println();
			System.out.println(datas);

			Cell[] rc = r.rawCells();
			for (Cell cell : rc) {
				data = cell.getFamilyArray();
				offset = cell.getFamilyOffset();
				length = cell.getFamilyLength();
				datas = new String(data, offset, length);
				System.out.print(datas + ":");

				data = cell.getQualifierArray();
				offset = cell.getQualifierOffset();
				length = cell.getQualifierLength();
				datas = new String(data, offset, length);
				System.out.print(datas + ":");

				data = cell.getValueArray();
				offset = cell.getValueOffset();
				length = cell.getValueLength();
				datas = new String(data, offset, length);
				System.out.println(datas);

			}
		}
		table.close();
	}

	public void viewTable(String tableName, String family, String qualifier,
			String value) throws IOException {
		HTable htable = new HTable(config, tableName);
		Filter filter = new SingleColumnValueFilter(Bytes.toBytes(family),
				Bytes.toBytes(qualifier), CompareOp.EQUAL, Bytes.toBytes(value));
		Scan scan = new Scan();
		scan.setFilter(filter);
		scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
		ResultScanner scanner = htable.getScanner(scan);
		for (Result r : scanner) {
			byte[] data = r.getRow();
			int offset = 0;
			int length = 0;
			String datas = new String(data);
			System.out.println();
			System.out.println(datas);

			Cell[] rc = r.rawCells();
			for (Cell cell : rc) {
				data = cell.getFamilyArray();
				offset = cell.getFamilyOffset();
				length = cell.getFamilyLength();
				datas = new String(data, offset, length);
				System.out.print(datas + ":");

				data = cell.getQualifierArray();
				offset = cell.getQualifierOffset();
				length = cell.getQualifierLength();
				datas = new String(data, offset, length);
				System.out.print(datas + ":");

				data = cell.getValueArray();
				offset = cell.getValueOffset();
				length = cell.getValueLength();
				datas = new String(data, offset, length);
				System.out.println(datas);

			}
		}
		htable.close();
	}

	public void upsertRow(String tableName, String row, String columnFamily,
			String column, String value) throws IOException {
		HTable table = new HTable(config, tableName);
		Put put = new Put(Bytes.toBytes(row));
		put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column),
				Bytes.toBytes(value));
		table.put(put);
		table.close();
		System.out.println("The data has been inserted successfully!");
	}

	public void deleteRow(String tableName, String rowkey[]) throws IOException {
		HTable h = new HTable(config, tableName);
		List<Delete> list = new ArrayList<Delete>();
		for (String k : rowkey) {
			Delete d = new Delete(Bytes.toBytes(k));
			list.add(d);
		}
		h.delete(list);
		h.close();
	}

	public void getOne(String tableName, String rowkey, String columnFamily)
			throws Exception {
		HTable h = new HTable(config, tableName);
		Get g = new Get(Bytes.toBytes(rowkey));
		g.addFamily(Bytes.toBytes(columnFamily));
		Result r = h.get(g);
		Cell[] rc = r.rawCells();
		byte[] data = r.getRow();
		int offset = 0;
		int length = 0;
		String datas = new String(data);
		System.out.println();
		System.out.println(datas);
		for (Cell cell : rc) {
			data = cell.getFamilyArray();
			offset = cell.getFamilyOffset();
			length = cell.getFamilyLength();
			datas = new String(data, offset, length);
			System.out.print(datas + ":");

			data = cell.getQualifierArray();
			offset = cell.getQualifierOffset();
			length = cell.getQualifierLength();
			datas = new String(data, offset, length);
			System.out.print(datas + ":");

			data = cell.getValueArray();
			offset = cell.getValueOffset();
			length = cell.getValueLength();
			datas = new String(data, offset, length);
			System.out.println(datas);

		}
		h.close();
	}

	public void createTable(String tableName, String familys[])
			throws Exception {

		HBaseAdmin admin = new HBaseAdmin(config); // 客户端管理工具类
		if (admin.tableExists(tableName)) {
			admin.disableTable(tableName);
			admin.deleteTable(tableName);
			System.out.println("delete table " + tableName + " ok.");
		}
		HTableDescriptor tableDesc = new HTableDescriptor(
				TableName.valueOf(tableName));
		for (int i = 0; i < familys.length; i++) {
			tableDesc.addFamily(new HColumnDescriptor(familys[i]));
		}
		admin.createTable(tableDesc);
		admin.close();
		System.out.println("create table " + tableName + " ok.");
	}

	public void deleteTable(String tableName) throws Exception {
		HBaseAdmin admin = new HBaseAdmin(config);
		admin.disableTable(tableName);
		admin.deleteTable(tableName);
		System.out.println("delete table " + tableName + " ok.");
		admin.close();
	}

	public static void main(String[] args) {
		try {
			String[] familys = { "familys" };
			String tableName = "test1";
			Hbase hbase = new Hbase();
			hbase.createTable(tableName, familys);
			hbase.upsertRow(tableName, "row", "familys", "column", "value1");
			hbase.upsertRow(tableName, "row", "familys", "column2", "value1");
			hbase.upsertRow(tableName, "row3", "familys", "column2", "value");
			System.out.println("====================");
			// hbase.viewTable(tableName, "", "", "value1");
			List<String[]> arr = new ArrayList<String[]>();
			String[] s = { "familys", "column2", "value1" };
			arr.add(s);
			hbase.selectByFilter(tableName, arr);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
